package com.ruoyi.flink.chapten5;

import java.util.ArrayList;
import java.util.Properties;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

public class SourceTest {

	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		env.setParallelism(1);
		// 1: 从文件中读取数据
		DataStreamSource<String> stream1 = env.readTextFile("input/clicks.txt");
		// 从集合中读取数据
		ArrayList<Integer> nums = new ArrayList<Integer>();
		nums.add(2);
		nums.add(5);
		DataStreamSource<Integer> numStream = env.fromCollection(nums);
		ArrayList<Event> events = new ArrayList<Event>();
		events.add(new Event("Mary", "./home", 1000L));
		events.add(new Event("Bob", "./cart", 2000L));
		DataStreamSource<Event> stream2 = env.fromCollection(events);
		// 从元素中读取数据
		DataStreamSource<Event> stream3 = env.fromElements(
				new Event("Mary", "./home", 1000L),
				new Event("Bob", "./cart", 2000L)
	    );
        // 4. 从Socket文本流读取
        DataStreamSource<String> stream4 = env.socketTextStream("localhost", 7777);
   
//		stream1.print("1");
//		numStream.print("nums");
//		stream2.print("2");
//		stream3.print("3");
//		stream4.print("4");
		
		// 5. 从kafka中获取数据
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.1.125:9092");
        properties.setProperty("group.id", "consumer-group");
        properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("auto.offset.reset", "latest");

        DataStreamSource<String> stream = env.addSource(new FlinkKafkaConsumer<String>(
                "clicks",
                new SimpleStringSchema(),
                properties
        ));


        stream.print("Kafka");
		
		
		
		
		
		env.execute();
         

	}

}
